package studio.raptor.dispatcher.scheduler.zookeeper;

import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import javax.annotation.PreDestroy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import studio.raptor.dispatcher.scheduler.base.CoordinatorRegistryCenter;
import studio.raptor.dispatcher.scheduler.exception.RegExceptionHandler;

/**
 * 基于Zookeeper的注册中心.
 *
 * @author jack
 * @since 1.0
 */
@Component
public final class ZookeeperRegistryCenter implements CoordinatorRegistryCenter {

  private static final Logger log = LoggerFactory.getLogger(ZookeeperRegistryCenter.class);


  @Autowired
  ZookeeperConfiguration zkConfig;

  private CuratorFramework client;


  @Override
  public void init() {
    log.debug("zookeeper registry center init, server lists is: {}.", zkConfig.getServers());
    CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
        .connectString(zkConfig.getServers())
        .retryPolicy(new ExponentialBackoffRetry(zkConfig.getBaseSleepTimeMilliseconds(),
            zkConfig.getMaxRetries(), zkConfig.getMaxSleepTimeMilliseconds()))
        .namespace(zkConfig.getNamespace());
    if (0 != zkConfig.getSessionTimeoutMilliseconds()) {
      builder.sessionTimeoutMs(zkConfig.getSessionTimeoutMilliseconds());
    }
    if (0 != zkConfig.getConnectionTimeoutMilliseconds()) {
      builder.connectionTimeoutMs(zkConfig.getConnectionTimeoutMilliseconds());
    }
    if (!Strings.isNullOrEmpty(zkConfig.getDigest())) {
      builder.authorization("digest", zkConfig.getDigest().getBytes(Charsets.UTF_8))
          .aclProvider(new ACLProvider() {

            @Override
            public List<ACL> getDefaultAcl() {
              return ZooDefs.Ids.CREATOR_ALL_ACL;
            }

            @Override
            public List<ACL> getAclForPath(final String path) {
              return ZooDefs.Ids.CREATOR_ALL_ACL;
            }
          });
    }
    client = builder.build();
    client.start();
    try {
      if (!client
          .blockUntilConnected(zkConfig.getMaxSleepTimeMilliseconds() * zkConfig.getMaxRetries(),
              TimeUnit.MILLISECONDS)) {
        client.close();
        throw new KeeperException.OperationTimeoutException();
      }
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
    }


  }

  @Override
  public void close() {
    waitForCacheClose();
    CloseableUtils.closeQuietly(client);
  }


  @Override
  public String getDirectly(final String key) {
    try {
      return new String(client.getData().forPath(key), Charsets.UTF_8);
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
      return null;
    }
  }

  @Override
  public List<String> getChildrenKeys(final String key) {
    try {
      List<String> result = client.getChildren().forPath(key);
      Collections.sort(result, new Comparator<String>() {

        @Override
        public int compare(final String o1, final String o2) {
          return o2.compareTo(o1);
        }
      });
      return result;
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
      return Collections.emptyList();
    }
  }

  @Override
  public int getNumChildren(final String key) {
    try {
      Stat stat = client.checkExists().forPath(key);
      if (null != stat) {
        return stat.getNumChildren();
      }
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
    }
    return 0;
  }

  @Override
  public boolean isExisted(final String key) {
    try {
      return null != client.checkExists().forPath(key);
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
      return false;
    }
  }

  @Override
  public void persist(final String key, final String value) {
    try {
      if (!isExisted(key)) {
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
            .forPath(key, value.getBytes(Charsets.UTF_8));
      } else {
        update(key, value);
      }
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
    }
  }

  @Override
  public void update(final String key, final String value) {
    try {
      client.inTransaction().check().forPath(key).and().setData()
          .forPath(key, value.getBytes(Charsets.UTF_8)).and().commit();
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
    }
  }

  @Override
  public void persistEphemeral(final String key, final String value) {
    try {
      if (isExisted(key)) {
        client.delete().deletingChildrenIfNeeded().forPath(key);
      }
      client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
          .forPath(key, value.getBytes(Charsets.UTF_8));
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
    }
  }

  @Override
  public String persistSequential(final String key, final String value) {
    try {
      return client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL)
          .forPath(key, value.getBytes(Charsets.UTF_8));
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
    }
    return null;
  }

  @Override
  public void persistEphemeralSequential(final String key) {
    try {
      client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
          .forPath(key);
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
    }
  }

  @Override
  public void remove(final String key) {
    try {
      client.delete().deletingChildrenIfNeeded().forPath(key);
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
    }
  }

  @Override
  public long getRegistryCenterTime(final String key) {
    long result = 0L;
    try {
      persist(key, "");
      result = client.checkExists().forPath(key).getMtime();
    } catch (final Exception ex) {
      RegExceptionHandler.handleException(ex);
    }
    Preconditions.checkState(0L != result, "Cannot get registry center time.");
    return result;
  }


  public ZookeeperConfiguration getZkConfig() {
    return zkConfig;
  }

  public CuratorFramework getClient() {
    return client;
  }

  /* 因为异步处理, 可能会导致client先关闭而cache还未关闭结束.
    * 等待Curator新版本解决这个bug.
    * BUG地址：https://issues.apache.org/jira/browse/CURATOR-157
     */
  private void waitForCacheClose() {
    try {
      Thread.sleep(500L);
    } catch (final InterruptedException ex) {
      Thread.currentThread().interrupt();
    }
  }

}
